package com.chencong.online.function;

import com.chencong.online.bean.AdClinkBehaviorBean;
import com.chencong.online.bean.BlackListWarning;
import com.chencong.online.utils.TimestampTransformUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.junit.Test;

/**
 * @program: user-behavior-analysis-online
 * @ClassName BlockListProcessFunc
 * @description:用户对同一广告点击超过上限输出侧输出流报警
 * @author: chencong
 * @create: 2021-12-29 14:12
 **/
public class BlackListProcessFunc extends KeyedProcessFunction<Tuple, AdClinkBehaviorBean, AdClinkBehaviorBean> {
    // 定义属性：点击次数上限
    private Integer countUpperBound;

    public BlackListProcessFunc(Integer countUpperBound) {
        this.countUpperBound = countUpperBound;
    }

    // 定义状态，保存当前用户对某一广告的点击次数
    ValueState<Long> countState;

    // 定义一个标志状态，保存当前用户是否已经被发送到了黑名单里
    ValueState<Boolean> isBlackListState;

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化点击次数状态
        countState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("count-state", Long.class, 0L));
        //初始化黑名单状态
        isBlackListState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-block-list-state", Boolean.class, false));

    }

    @Override
    public void processElement(AdClinkBehaviorBean value, KeyedProcessFunction<Tuple, AdClinkBehaviorBean, AdClinkBehaviorBean>.Context ctx, Collector<AdClinkBehaviorBean> out) throws Exception {
        // 判断当前用户对同一广告的点击次数，如果不够上限，就count加1正常输出；如果达到上限，直接过滤掉，并侧输出流输出黑名单报警
        // 首先获取当前的count值
        Long adClinkCount = countState.value();
        // 1. 判断是否是第一个数据，如果是的话，注册一个第二天0点的定时器(这里用处理时间)
        if (adClinkCount == 0L) {
            Long ts = (ctx.timerService().currentProcessingTime() / (24 * 60 * 60 * 1000) + 1) * (24 * 60 * 60 * 1000) - 8 * 60 * 60 * 1000;
            System.out.println(TimestampTransformUtil.milliTimestampToLocalDateTime(ts));
            ctx.timerService().registerProcessingTimeTimer(ts);
        }
        // 2. 对于超过上限
        if (adClinkCount >= countUpperBound) {
            // 判断是否输出到黑名单过，如果没有的话就输出到侧输出流
            Boolean isBlackList = isBlackListState.value();
            if (!isBlackList) {
                //如果没有。放入侧输出流黑名单
                BlackListWarning blackListWarning = new BlackListWarning(value.getUserId(), value.getAdId(), "点击量超过：" + adClinkCount + "次");

                ctx.output(new OutputTag<BlackListWarning>("blackList") {
                }, blackListWarning);
                //更新状态
                isBlackListState.update(true);
            }

            //如果超过上限并且已经报过警时，不再执行下面操作
            return;
        }
        //3.没有超过上限
        // 点击次数状态加1，更新状态，正常输出当前数据到主流
        countState.update(adClinkCount + 1);
        out.collect(value);
    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<Tuple, AdClinkBehaviorBean, AdClinkBehaviorBean>.OnTimerContext ctx, Collector<AdClinkBehaviorBean> out) throws Exception {
        //第二天凌晨点，定时器触发，情况状态
        countState.clear();
        isBlackListState.clear();
    }
}
